package com.navinfo.opentsp.platform.computing.parallel.application;

import com.alibaba.fastjson.JSON;
import com.google.common.base.Strings;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
import com.navinfo.opentsp.platform.computing.parallel.util.PropertiesUtil;

import com.navinfo.platform.bigdata.dto.VehicleStatusAdditionDto;
import com.navinfo.stream.qingqi.protocol.java.LCLocationData;
import com.navinfo.stream.qingqi.protocol.java.LCStatusType.StatusType;
import com.navinfo.stream.qingqi.protocol.java.LCVehicleStatusData;
import org.apache.commons.lang3.StringUtils;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.broadcast.Broadcast;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.SQLContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import scala.Tuple2;

import java.io.*;
import java.sql.*;
import java.util.*;
import java.text.SimpleDateFormat;

/**
 * 从HDFS导出LocationData数据到CSV中
 * Created by web on 2018/11/22.
 */
public class ExportHdfsLocationData {
    private static Logger logger = LoggerFactory.getLogger(ExportHdfsLocationData.class);
    private static final int MAX_HEIGHT = 4000;
    private static double EARTH_RADIUS = 6378.137D;
    private static final StatusType[] statusTypeArray = {StatusType.oilValue, StatusType.rotation, StatusType.coolingWaterTem, StatusType.totalFuelConsumption, StatusType.fuelConsumptionRate, StatusType.engineTorMode, StatusType.driverEnginePercentTor, StatusType.actualEnginePercentTor, StatusType.accPedalPos, StatusType.nominalFrictionPercentTrq, StatusType.parkingBrakeSwitch, StatusType.wheelBasedVehicleSpd, StatusType.cruiseCtrlActive, StatusType.brakeSwitch, StatusType.clutchSwitch, StatusType.cruiseCtrlCoastSwitch, StatusType.cruiseCtrlResumeSwitch, StatusType.cruiseCtrlAccSwitch, StatusType.cruiseCtrlSetSpd, StatusType.cruiseCtrlStates, StatusType.outPutRoateSpeed};

    public static void main(String[] args) {
        long startTime = System.currentTimeMillis();
        logger.info("开始执行ExportHdfsLocationData操作......");

        String mater = null;
        String dataTime = "";
        String csvPath = "";
        if (args == null || args.length == 0) {
            dataTime = "20181122";
            mater = "local[*]";
            csvPath = "D:/";
        } else {
            dataTime = args[0];
            csvPath = args[1];
        }
        logger.info("执行参数dataTime={}", dataTime);
        logger.info("执行参数csvPath={}", csvPath);

        //公共数据
        Map<String, String> commonData = new HashMap<>();
        commonData.put("csvPath", csvPath);
        commonData.put("dataTime", dataTime);
        commonData.put("fieldUnitKey", PropertiesUtil.getProperties("fieldUnitKey"));
        commonData.put("fieldUnitValue", PropertiesUtil.getProperties("fieldUnitValue"));
        commonData.put("headerField", PropertiesUtil.getProperties("headerField"));

        //车辆配置数据
        Map<String, List<String>> vehicleInfoCache = loadVehicleInfo(commonData);

        JavaSparkContext jsc = null;
        try {
            SparkConf conf = new SparkConf().setAppName("ExportHdfsLocationData");
            if (StringUtils.isNotBlank(mater)) {
                conf.setMaster(mater);
            }
            jsc = new JavaSparkContext(conf);
            SQLContext sqlContext = new SQLContext(jsc);

            DataFrame parquetDF = sqlContext.read().parquet(String.format(PropertiesUtil.getProperties("hdfs.0200.path"), dataTime));
            logger.info("load data from parquet, time={}", System.currentTimeMillis() - startTime);

            JavaRDD<Row> rowRDD = parquetDF.select("tid", "status", "longitude", "latitude", "height", "speed", "gpsdate", "mileage", "statusaddition").javaRDD();

            JavaPairRDD<String, Iterable<Row>> groupRDD = rowRDD.groupBy(row -> row.<String>getAs("tid"));

//            int partitionsNum = groupRDD.partitions().size();
//            commonData.put("partitionsNum", String.valueOf(partitionsNum));

            final Broadcast<Map<String, List<String>>> vehicleBroadcast = jsc.broadcast(vehicleInfoCache);
            final Broadcast<Map<String, String>> commonBroadcast = jsc.broadcast(commonData);
            logger.info("广播变量1大小:{}", vehicleInfoCache.size());
            logger.info("广播变量2大小:{}", commonData.size());

            groupRDD.foreachPartition(tuple2Iterator -> {
                Map<String, LastCacheInfo> lastCache = new HashMap();

                while (tuple2Iterator.hasNext()) {
                    Tuple2<String, Iterable<Row>> data = tuple2Iterator.next();
                    String tid = data._1;
                    List<String> vehicleInfo = vehicleBroadcast.getValue().get(tid);
                    if (vehicleInfo == null) {
                        continue;
                    }

                    String vin = commonBroadcast.getValue().get(tid);
                    String path = commonBroadcast.getValue().get("csvPath");
                    String date = commonBroadcast.getValue().get("dataTime");

                    List<String> saveData = new ArrayList<String>();
                    saveData.addAll(vehicleInfo);
                    saveData.add("数据生成日期:" + date.substring(0, 4) + "/" + date.substring(4, 6) + "/" + date.substring(6, 8) + "/" + "00/00");
                    saveData.add(commonBroadcast.getValue().get("fieldUnitKey"));
                    saveData.add(commonBroadcast.getValue().get("fieldUnitValue"));
                    saveData.add(commonBroadcast.getValue().get("headerField"));

                    //通过gpsdata排序，并去重
                    TreeSet<Row> rowSet = Sets.newTreeSet(Comparator.comparingInt((Row o) -> Integer.parseInt(o.getAs("gpsdate"))));
                    data._2.forEach(e -> rowSet.add(e));

                    SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy\\MM\\dd\\HH\\mm\\ss");
                    rowSet.forEach(row -> {
                        //"tid", "status", "longitude", "latitude", "height", "speed", "gpsdate", "mileage", "statusaddition"
                        int status = (Long.parseLong(row.getAs(1)) & 1L) == 1L ? 1 : 0;
                        long longitude = Long.parseLong(row.getAs("longitude"));
                        long latitude = Long.parseLong(row.getAs("latitude"));
                        int height = Integer.parseInt(row.getAs("height"));

                        height = getHeight(lastCache, vin, latitude, longitude, height);
                        Map<StatusType, Object> statusDataMap = getStatusDataMap(row.getAs("statusaddition"));
                        StringBuilder fieldValue = new StringBuilder("");
                        fieldValue.append(dateFormat.format(Long.parseLong(row.getAs("gpsdate")) * 1000L)).append(",")
                                .append(longitude).append(",")
                                .append(latitude).append(",")
                                .append(height).append(",")
                                .append(Integer.parseInt(row.getAs("speed"))).append(",")
                                .append(status).append(",")
                                .append(Objects.toString(statusDataMap.get(StatusType.mileage),"0")).append(",")
                                .append(Long.parseLong(row.getAs("mileage")) / 1000.0D).append(",");
                        for (StatusType statusType : statusTypeArray) {
                            fieldValue.append(Objects.toString(statusDataMap.get(statusType),"0")).append(",");
                        }
                        fieldValue.append("0");

                        saveData.add(fieldValue.toString());
                    });

                    writeDataToCsv(saveData, path, vin + date + ".csv");
                }
            });
        } finally {
            logger.info("结束ExportHdfsLocationData操作，耗时={}" , System.currentTimeMillis()-startTime);
            if (null != jsc) {
                jsc.close();
            }
        }
    }


    /**
     * 加载车辆配置信息
     * @param commonData
     */
    private static Map<String, List<String>> loadVehicleInfo(Map<String, String> commonData) {
        Map<String, List<String>> vehicleInfo = new HashMap<>();

        String driver = "com.mysql.jdbc.Driver"; // URL指向要访问的数据库名game
        String url = PropertiesUtil.getProperties("mysql.url"); // MySQL配置时的用户名
        String user = PropertiesUtil.getProperties("mysql.username"); // MySQL配置时的密码
        String password = PropertiesUtil.getProperties("mysql.password");

        long startTime = System.currentTimeMillis();
        try (
                Connection conn = DriverManager.getConnection(url, user, password);
                PreparedStatement ps = conn.prepareStatement(PropertiesUtil.getProperties("vehicle.sql"));
                ResultSet rs = ps.executeQuery();
        ) {
            Class.forName(driver);
            ResultSetMetaData metaData = ps.getMetaData();

            while (rs.next()) {
                String tid = rs.getString(metaData.getColumnLabel(18));
                if (Strings.isNullOrEmpty(tid)) {
                    continue;
                }

                /*底盘号*/
                String chassisNum = getString(rs.getString(metaData.getColumnLabel(1)), "底盘号");
                String engineType = getString(rs.getString(metaData.getColumnLabel(2)), "发动机类型");
                String oriVin = rs.getString(metaData.getColumnLabel(3));
                String vin = getString(oriVin, "VIN");
                String driveForm = getString(rs.getString(metaData.getColumnLabel(4)), "驱动形式");
                String carType = getString(rs.getString(metaData.getColumnLabel(5)), "车辆类别");
                String platType = getString(rs.getString(metaData.getColumnLabel(6)), "平台车型");
                String productTime = getString(rs.getString(metaData.getColumnLabel(7)), "车辆生产日期");
                String teamName = getString(rs.getString(metaData.getColumnLabel(9)), "订单信息");
                String lhq = getString(rs.getString(metaData.getColumnLabel(10)), "离合器");
                String bsx = getString(rs.getString(metaData.getColumnLabel(11)), "变速箱类型");
                String hq = getString(rs.getString(metaData.getColumnLabel(12)), "后桥类型");
                String cj = getString(rs.getString(metaData.getColumnLabel(13)), "车架类型");
                String zj = getString(rs.getString(metaData.getColumnLabel(14)), "轴距");
                String qt = getString(rs.getString(metaData.getColumnLabel(15)), "其他");
                String lt = getString(rs.getString(metaData.getColumnLabel(16)), "轮胎类型");

                //tid与车辆信息绑定
                vehicleInfo.put(tid, Lists.newArrayList(vin, chassisNum, platType, carType, engineType, lhq, driveForm, bsx, hq, cj, zj, lt, teamName, qt, productTime));
                commonData.put(tid, oriVin);
            }
        } catch (Exception e) {
            logger.error("加载车辆配置信息失败......", e);
        } finally {
            logger.info("加载车辆配置信息完成，耗时={}，大小={}", System.currentTimeMillis() - startTime, vehicleInfo.size());
            return vehicleInfo;
        }
    }

    private static String getString(String data, String prefix) {
        return prefix + ":" + Objects.toString(data, "");
    }

    private static Map<StatusType, Object> getStatusDataMap(String statusaddition) {
        LCLocationData.VehicleStatusAddition.Builder builder = LCLocationData.VehicleStatusAddition.newBuilder();
        JSON.parseArray(statusaddition, String.class).forEach(e -> {
            builder.addStatus(VehicleStatusAdditionDto.parsePB(e));
        });

        Map<StatusType, Object> statusDataMap = new HashMap<>();
        for (LCVehicleStatusData.VehicleStatusData vehStatus : builder.getStatusList()) {
            StatusType types = vehStatus.getTypes();
            long statusValue = vehStatus.getStatusValue();

            if (StatusType.fuelConsumptionRate.equals(types)) {
                statusDataMap.put(types, (float)statusValue / 100.0F / 1000.0F);
                continue;
            }

            statusDataMap.put(types, statusValue/ 100L);
        }
        return statusDataMap;
    }

    private static int getHeight(Map<String, LastCacheInfo> lastCache, String vin, long lat, long lng, int height) {
        int height0 = 0;
        if (!lastCache.containsKey(vin)) {
            LastCacheInfo lastInfo = new LastCacheInfo();
            lastInfo.setHeight(height > MAX_HEIGHT ? 500 : height);
            lastInfo.setLat(lat);
            lastInfo.setLng(lng);
            lastCache.put(vin, lastInfo);
        } else {
            LastCacheInfo lastInfo = lastCache.get(vin);
            boolean flag = validateDistance(lastInfo, lat, lng, height);
            if (!flag) {
                height0 = lastInfo.getHeight();
            } else {
                height0 = height;
                lastInfo.setHeight(height);
            }
            lastInfo.setLat(lat);
            lastInfo.setLng(lng);
        }
        return height0;
    }

    private static void writeDataToCsv(List<String> dataList, String path, String fileName) {
        if (dataList == null) {
            return;
        }
        File file = new File(path);
        if (!file.exists()) {
            file.mkdirs();
        }

        try (
            BufferedWriter writer = new BufferedWriter(new FileWriter(path + File.separator + fileName));
        ){
            for (String lineData : dataList) {
                writer.write(lineData + System.lineSeparator());
            }

            writer.flush();
        } catch (IOException e) {
            logger.error("写入csv文件["+path + File.separator + fileName+"]失败......", e);
        }
    }


    private static double rad(double d) {
        return d * 3.141592653589793D / 180.0D;
    }

    private static double getDistance(double lat1, double lng1, double lat2, double lng2) {
        double radLat1 = rad(lat1);
        double radLat2 = rad(lat2);
        double a = Math.abs(radLat1 - radLat2);
        double b = Math.abs(rad(lng1) - rad(lng2));

        double s = 2.0D * Math.asin(Math.sqrt(Math.pow(Math.sin(a / 2.0D), 2.0D) + Math.cos(radLat1) * Math.cos(radLat2) * Math.pow(Math.sin(b / 2.0D), 2.0D)));
        s *= EARTH_RADIUS;
        return Math.round(s * 1000.0D);
    }

    private static boolean validateDistance(LastCacheInfo info, long lat, long lng, int height) {
        double distance = getDistance(info.getLat(), info.getLng(), lat, lng);
        int ht = Math.abs(info.getHeight() - height);
        if (distance >= 2 * ht) {
            return true;
        }
        return false;
    }

    private static class LastCacheInfo implements Serializable {
        private long lat;
        private long lng;
        private int height;
        long getLat() {
            return this.lat;
        }
        void setLat(long lat) {
            this.lat = lat;
        }
        long getLng() {
            return this.lng;
        }
        void setLng(long lng) {
            this.lng = lng;
        }
        int getHeight() {
            return this.height;
        }
        void setHeight(int height) {
            this.height = height;
        }
    }
}
